package org.example.rocketmq.msg_type.broadcast;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 广播消息并没有特定的消息消费者样例，这是因为涉及到消费者的集群消费模式
 * MessageModel.BROADCASTING: 广播消息。一条消息会发给所有订阅了 对应主题的消费者，不管消费者是不是同一个消费组
 * MessageModel.CLUSTERING: 集群消息，每一条消息只会被同一个消费者组中的一个实例消费。
 */
public class ConsumerBroadcast {
    public static void main(String[] args) throws Exception {
        consumer();
    }

    /**
     * 顺序消息值生产者局部有序发送到一个 queue, 但多个 queue 之间是全局无序的。
     * 顺序消息消费者样例： 通过 MessageListenerOrderly 消费者每次读取消息都只从一个 queue 中获取（通过加锁的方式实现）
     */
    public static void consumer() throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_order_group_01");
        consumer.setNamesrvAddr("rocketmq.jun.cn:9876");
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.subscribe("TopicTest", "*");
        consumer.setMessageModel(MessageModel.BROADCASTING);

        consumer.registerMessageListener(new MessageListenerOrderly() { //  顺序消费
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for (int i = 0; i < list.size(); i++) {
                    System.out.println(i + "_消息消费成功_" + new String(list.get(i).getBody()));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });


        /*consumer.registerMessageListener(new MessageListenerConcurrently() { // 并发消费
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (int i = 0; i < list.size(); i++) {
                    System.out.println(i + "_消息消费成功_" + new String(list.get(i).getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });*/

        consumer.start();
        System.out.printf("Consumer started %n");
    }
}
